package org.poem.util;

import org.apache.kafka.clients.producer.Partitioner;
import org.apache.kafka.common.Cluster;

import java.util.Map;

/**
 * 自定义分区器
 *
 * @author Administrator
 */
public class PartitionUtil implements Partitioner {

    /**
     * 分区数
     */
    private static final Integer PARTITION_NUM = 3;

    @Override
    public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
        if (null == key) {
            return 0;
        }
        // key取模
        int hash = String.valueOf(value).hashCode() + key.toString().hashCode();
        int partition = Math.abs((int) (hash % PARTITION_NUM));
        return partition;
    }

    @Override
    public void close() {

    }

    @Override
    public void configure(Map<String, ?> configs) {

    }
}